package edu.nepu.flink.api.watermark;

import edu.nepu.flink.api.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * @Date 2024/3/1 14:54
 * @Created by chenshuaijun
 */
public class WatermarkGeneStrategy {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> streamSource = env.socketTextStream("hadoop102", 9999).map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(" ");
                return new WaterSensor(split[0],Long.parseLong(split[1]),Integer.parseInt(split[2]));
            }
        });
        /**
         * TODO 下面是关于waterMark的解读
         * 1、什么是waterMark
         *      waterMark是用来衡量事件时间进展的。
         * 2、waterMark的特性
         *     (1) watermark是一插入数据流中的一个标志，可以认为是一个特殊的数据
         *     (2) watermark的主要内容是一个时间戳，用来表示时间的进展
         *     (3) watermark是基于数据中的时间戳生成的
         *     (4) watermark的时间戳必须是单调不减的，用来保证任务的事件时间时钟一直是向前进展的
         *     (5) watermark可以设置延迟时间，来保证乱序数据的正确处理
         *     (6) watermark(t)表示当前的事件时间已经到达了t，这就表示t之前的数据已经全部到达
         * 3、waterMark的生成方式
         *     (1) 周期性生成，默认的周期是200ms
         *     (2) 间歇性的生成，没来一条数据就生成一次waterMark
         *          这样虽然延迟小，但是会严重的影响效率
         */
        SingleOutputStreamOperator<WaterSensor> watermarkStream = streamSource.assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                return element.getTs() * 1000;
            }
        }).withIdleness(Duration.ofSeconds(2)));

        watermarkStream.keyBy(WaterSensor::getId)
                       .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                       .reduce(new ReduceFunction<WaterSensor>() {
                           @Override
                           public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                               return new WaterSensor(value1.id, value2.ts, value1.vc + value2.vc);
                           }
                       }).print();
    }
}
